package org.luxor.commons.redis.client;

import org.luxor.commons.redis.component.RedisListenerReceiver;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.util.Assert;

/**
 * Redis 发布与订阅（非持久化）
 *
 * 注意：客户端订阅channel后只能收到后续发布到该频道的消息，之前发送的不会缓存。
 *
 * @author WangFan
 * @author Mr.Yan  @date 2019/10/8
 * @version 1.1 (GitHub文档: https://github.com/whvcse/RedisUtil )
 */
public class RedisTopicClient {

    private final StringRedisTemplate redisTemplate;

    private final RedisMessageListenerContainer redisMessageListenerContainer;

    public RedisTopicClient(StringRedisTemplate redisTemplate, RedisMessageListenerContainer redisMessageListenerContainer) {
        Assert.notNull(redisTemplate, "stringRedisTemplate must be specified");
        this.redisTemplate = redisTemplate;

        Assert.notNull(redisMessageListenerContainer, "RedisMessageListenerContainer must be specified");
        this.redisMessageListenerContainer = redisMessageListenerContainer;
    }

    /**
     * 订阅某一个主题（会阻塞，非持久化订阅）
     *
     * @param redisListener 回调方法
     * @param topic         主题(频道)名
     */
    public void subscribe(RedisListenerReceiver redisListener, Topic topic) {
        Assert.notNull(topic, "non null topic required");
        Assert.notNull(redisListener, "non null redisListener required");

        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(redisListener, "onMessage");
        messageListenerAdapter.afterPropertiesSet();
        redisMessageListenerContainer.addMessageListener(messageListenerAdapter, topic);
    }

    /**
     * 将消息推送到指定主题
     *
     * @param message 消息内容
     * @param topic   主题(频道)名
     */
    public void publish(String message, Topic topic) {
        Assert.notNull(topic, "non null topic required");
        redisTemplate.convertAndSend(topic.getTopic(), message);
    }

}
